package sunyu.kit.kafka;

import org.apache.kafka.clients.producer.Producer;
import org.slf4j.Logger;
import sunyu.kit.common.LogKit;

import java.util.Properties;

/**
 * 生产者模版
 *
 * @author 孙宇
 */
public class KafkaProducer {

    private static final Logger logger = LogKit.getLogger();

    private Producer<String, String> producer;

    public KafkaProducer(String servers, String keySerializer, String valueSerializer) {
        logger.info("初始化Kafka Producer开始");
        logger.info("servers {}", servers);
        logger.info("keySerializer {}", keySerializer);
        logger.info("valueSerializer {}", valueSerializer);

        Properties props = new Properties();
        props.put("bootstrap.servers", servers);
        props.put("key.serializer", keySerializer);
        props.put("value.serializer", valueSerializer);
        /*props.put("acks", "all");
        props.put("retries ", 1);
        props.put("buffer.memory", 33554432);
        */
        producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);

        logger.info("初始化Kafka Producer结束");
    }

    public Producer<String, String> getProducer() {
        return producer;
    }

}
